package com.gss.plugin.kafka.config;

import com.gss.plugin.kafka.util.IpAddressUtil;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @author: DYang
 * @since: 2023/8/24
 * @description:
 */
@Data
@Configuration
@ConfigurationProperties(prefix = "stars.kafka")
public class KafkaProperties {

    /**
     * 服务地址,多个用“,”隔开
     */
    private String servers = "localhost:9092";

    /**
     * 是否开启安全
     */
    private boolean securityEnable = false;
    /**
     * kafka安全认证,SASL_PLAINTEXT
     */
    private String securityProtocol =  "SASL_PLAINTEXT";
    /**
     * PLAIN
     */
    private String saslMechanism =  "PLAIN";
    /**
     * 此处的配置和kafka中的加密配置保持一致
     * saslJaasConfig
     */
    private String saslJaasConfig = "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test\" password=\"test@123\";";


    /**
     * 生产者
     */
    private ProducerConfig producer = new ProducerConfig();
    /**
     * 消费者
     */
    private ConsumerConfig consumer = new ConsumerConfig();


    @Data
    public static class ProducerConfig{
        /**
         * 是否启用
         */
        private boolean enable = true;
        /**
         * topic 配置
         */
        private List<TopicConfig> topics;

        /**
         * ack
         * 0: 不需要服务端确认，即可再次提交,小概率会丢失数据
         * 1：需要服务端leader 确认，才能提交下一条，小概率follwer 宕机会丢失数据
         * -1: 需要等到follwer 确认，才能提交下一条，效率最低，但是最可靠
         */
        private String ack = "-1";

        /**
         * 重试次数
         */
        private int retries = 9999;
        /**
         * 最大请求数量
         */
        private int maxRequestSize = 10;
        /**
         * 批量存储
         */
        private long batchSize = 4096;
        /**
         * 数据库缓存
         */
        private long bufferMemory = 40960;
    }

    @Data
    public static class ConsumerConfig{

        /**
         * topic 信息
         */
        private String topics;
        /**
         * 消费者组
         */
        private String groupId = IpAddressUtil.getLocalIP();

        /**
         * 启动订阅模式
         */
        private boolean subscribeEnabled = false;

        /**
         * 启用listener 模式
         */
        private boolean listenerEnable = true;

        /**
         * 是否启用消费者
         */
        private boolean enable = true;
        /**
         * 是否自动提交消费偏移量
         */
        private boolean autoCommit = false;
        /**
         * 提交偏移量session 超时（ms）
         */
        private long sessionTimeout = 6000;
        /**
         * 偏移量提交频次
         */
        private int commitInterval = 100;
        /**
         * 消费偏移量方式
         * latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
         * earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
         */
        private String offsetReset = "latest";
        /**
         * 一个应用程序消费者线程数量，默认给3个
         */
        private int concurrency = 3;
        /**
         * 拉取数据消费请求超时时间(ms)
         */
        private long requestTimeout = 30000;
        /**
         * 最大请求拉取间隔时间（ms）
         */
        private long maxPollInterval = 600000;
        /**
         * 单次poll调用返回的最大消息数,条数据需要在在session.timeout.ms这个时间内处理完
         */
        private int maxPollRecords = 30;
        /**
         * 单次获取数据的最大消息数8M broker 配置
         */
        private long maxMessageBytes = 8388608;
        /**
         * 单次获取数据的最大消息数10M max.partition.fetch.bytes 会话6s内处理的最大数据量大于max.message.bytes
         */
        private long fetchMaxBytes = 10485760;
        /**
         * 心跳频次 大小不能超过session.timeout.ms
         */
        private long heartbeatInterval = 2000;
        /**
         * 轮训策略 partition.assignment.strategy: org.apache.kafka.clients.consumer.RoundRobinAssignor
         */
        private String partitionAssignmentStrategy = "org.apache.kafka.clients.consumer.StickyAssignor";

    }

    @Data
    public static class TopicConfig{
        /**
         * topic 名称
         */
        private String topic;
        /**
         * 消费者组
         */
        private String consumerGroup;


    }


}
